package com.cmcc.hackson3.main;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.RandomUtils;

import com.cmcc.hackson3.user.QueueManager;

public class Producer {
	private static final AtomicInteger  prefix = new AtomicInteger();
	private volatile boolean  started = false;
	private volatile boolean  terminal = false;
	private String name;
	
	public Producer(String name) {
		this.name = name + prefix.getAndIncrement();
	}
	
	public void stop() {
		terminal = true;
	}
	
	private String createPhone() {
		return String.valueOf(RandomUtils.nextLong(13800000000L, 13900000000L));
	}
	
	
	public void start() {
		if(started) return;
		Thread r = new Thread(null ,new Runnable() {

			@Override
			public void run() {
				while(!terminal) {
					try {
						for(int i = 0;i<RandomUtils.nextInt(1,6);i++) {
							Message msg = new Message(RandomUtils.nextInt(5, 60), createPhone());
							String topic = QueueManager.INS.chooseTopic(msg);
							LinkedBlockingQueue<Message> queue = RocketMQ.INS.getQueue(topic);
							queue.put(msg);
							QueueManager.INS.inqueue(topic, msg);
						}
						Thread.sleep((long)((40+(long)((Math.sin(System.currentTimeMillis()/1000%10000)+1)*100))*RandomUtils.nextFloat(0,1)));
					} catch (InterruptedException e) {
						
					}
				}
			}
			
			
		},name);
		r.setDaemon(true);
		r.start();
		QueueManager.INS.producerStarted(name);
	}

}
